package com.wudizaba.nb.kafkaProvider;

import config.kafkaConfig;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import javax.annotation.Resource;

@Component
@Slf4j
public class provider {


    @Resource
    private KafkaTemplate<String, Object> kafkaTemplete;

    public void sendMessage(String message) {
        if (kafkaTemplete == null) {
            throw new NullPointerException("kafkaTemplete is null");
        }
        log.info("准备发送消息:{}", message);
        ListenableFuture<SendResult<String, Object>> send = kafkaTemplete.send(kafkaConfig.TOPIC_TEST, message);
        send.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            //发送失败
            @Override
            public void onFailure(Throwable ex) {
                log.error("发送消息失败:{}", message);
                log.error("发送消息失败:{}", ex.getMessage());
            }

            //发送成功
            @Override
            public void onSuccess(SendResult<String, Object> result) {
                log.info("发送消息成功:{}", message);
                log.info("发送消息成功:{}", result.getRecordMetadata().toString());
            }
        });
    }


}
